/*
 * Copyright 2019-2025 the original author or authors.
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * https://www.apache.org/licenses/LICENSE-2.0
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.iiifi.kite.rocketmq.core;

import java.lang.annotation.Annotation;
import java.lang.reflect.Field;
import java.nio.charset.StandardCharsets;

import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.message.Message;

import com.iiifi.kite.core.util.JsonUtils;
import com.iiifi.kite.rocketmq.annotation.RocketMqKey;
import com.iiifi.kite.rocketmq.enums.DelayTimeLevel;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;

/**
 * RocketMQ的事务生产者的抽象基类
 *
 * @author kite@iiifi.com 花朝
 */
@Data
@Slf4j
public class MessageBuilder {

    private static final String[] DELAY_ARRAY = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h".split(" ");

    private String topic;
    private String tag;
    private String key;
    private Object message;
    private Integer delayTimeLevel;

    public static MessageBuilder of(String topic, String tag) {
        MessageBuilder builder = new MessageBuilder();
        builder.setTopic(topic);
        builder.setTag(tag);
        return builder;
    }

    public static MessageBuilder of(Object message) {
        MessageBuilder builder = new MessageBuilder();
        builder.setMessage(message);
        return builder;
    }

    public MessageBuilder topic(String topic) {
        this.topic = topic;
        return this;
    }

    public MessageBuilder tag(String tag) {
        this.tag = tag;
        return this;
    }

    public MessageBuilder key(String key) {
        this.key = key;
        return this;
    }

    public MessageBuilder delayTimeLevel(DelayTimeLevel delayTimeLevel) {
        this.delayTimeLevel = delayTimeLevel.getLevel();
        return this;
    }

    public Message build() {
        String messageKey = "";
        try {
            Field[] fields = message.getClass().getDeclaredFields();
            for (Field field : fields) {
                Annotation[] annotations = field.getAnnotations();
                if (annotations.length > 0) {
                    for (Annotation annotation : annotations) {
                        if (annotation.annotationType().equals(RocketMqKey.class)) {
                            field.setAccessible(true);
                            RocketMqKey mqKey = (RocketMqKey) annotation;
                            messageKey = StringUtils.isEmpty(mqKey.prefix())
                                    ? field.get(message).toString()
                                    : (mqKey.prefix() + field.get(message).toString());
                        }
                    }
                }
            }
        } catch (Exception e) {
            log.error("parse key error : {}", e.getMessage());
        }
        String str = JsonUtils.toJson(message);
        if (StringUtils.isEmpty(topic)) {
            if (StringUtils.isEmpty(getTopic())) {
                throw new RuntimeException("no topic defined to send this message");
            }
        }
        Message message = new Message(topic, str.getBytes(StandardCharsets.UTF_8));
        if (!StringUtils.isEmpty(tag)) {
            message.setTags(tag);
        }
        if (StringUtils.isNotEmpty(messageKey)) {
            message.setKeys(messageKey);
        }
        if (delayTimeLevel != null && delayTimeLevel > 0 && delayTimeLevel <= DELAY_ARRAY.length) {
            message.setDelayTimeLevel(delayTimeLevel);
        }
        return message;
    }
}
